package com.paas.form.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.codahale.metrics.ConsoleReporter;
import com.codahale.metrics.Counter;
import com.codahale.metrics.MetricRegistry;
import com.paas.common.util.AppUtil;
import com.paas.db.common.element.conf.etl.EtlConf;
import com.paas.etlx.core.etl.EtlPluginExecutor;
import com.paas.etlx.core.util.UuidUtil;
import com.paas.scheduler.exception.SchedulerException;
import com.paas.scheduler.model.JobInstance;
import com.paas.scheduler.spi.JobHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;

import java.util.Arrays;
import java.util.SortedMap;
import java.util.concurrent.TimeUnit;

@Slf4j
public class DistributionLogJobHandler implements JobHandler {
    private Environment environment;

    private JSONObject parameterObj;

    private JobInstance jobInstance;

    private String monitorColumnStr = "id,parent_id,distribution_title,distribution_type,description,operation_status," +
            "hand_out_type,job_id,data_id,create_id,create_org_id,create_date,modify_id,modify_date,status";
    private String monitorColumnLogStr = "id,parent_id,request_type,distribution_status,message,start_time,end_time,time_consuming," +
            "retry_count,request_parameter,interface_name,create_id,create_org_id,create_date,modify_id,modify_date,status";
    private String monitorTableName = "tools_realtime_distribution_monitoring";
    private String monitorLogTableName = "tools_realtime_distribution_monitoring_log";

    @Override
    public String getType() {
        return "MDM_TASK_LOG";
    }

    @Override
    public void init(JobInstance jobInstance) {
        initService();
        this.jobInstance = jobInstance;
        String jdbcUrl = environment.getProperty("spring.datasource.druid.first.url");
        String password = environment.getProperty("spring.datasource.druid.first.password");
        String username = environment.getProperty("spring.datasource.druid.first.username");
        if (parameterObj == null) {
            parameterObj = new JSONObject();
        }
        parameterObj.put("jdbcUrl", jdbcUrl);
        parameterObj.put("password", password);
        parameterObj.put("username", username);
    }

    @Override
    public void start() {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("writer", parseObj(monitorColumnStr, "writer", monitorTableName));
        jsonObject.put("reader", parseObj(monitorColumnStr, "reader", monitorTableName));
        executeEtl(jsonObject.toJSONString());
        JSONObject object = new JSONObject();
        object.put("writer", parseObj(monitorColumnLogStr, "writer", monitorLogTableName));
        object.put("reader", parseObj(monitorColumnLogStr, "reader", monitorLogTableName));
        executeEtl(object.toJSONString());
    }


    private JSONObject parseObj(String columnStr, String type, String tableName) {
        JSONObject jsonObject = new JSONObject();
        JSONObject parameter = parameterObj.clone();
        parameter.put("column", Arrays.asList(columnStr.split(",")));
        JSONArray writerArray = new JSONArray();
        JSONObject tableObj = new JSONObject();
        if ("writer".equals(type)) {
            tableObj.put("table", Arrays.asList(tableName + "_history"));
            parameter.put("postSql",Arrays.asList("delete from "+tableName));
            jsonObject.put("name", "mysqlwriter");
            parameter.put("writeMode", "replace");
        } else {
            tableObj.put("table", Arrays.asList(tableName));
            jsonObject.put("name", "mysqlreader");
        }
        writerArray.add(tableObj);
        parameter.put("connection", writerArray);
        jsonObject.put("parameter", parameter);
        return jsonObject;
    }

    private void executeEtl(String jsonStr) {
        EtlConf etlConf = JSON.parseObject(jsonStr, EtlConf.class);
        MetricRegistry registry = new MetricRegistry();
        String etlInstId = UuidUtil.shortUpperUuid();
        String pluginId = etlInstId + "-" + String.format("%04d", 1);
        ConsoleReporter reporter = null;
        EtlPluginExecutor etlPluginExecutor = new EtlPluginExecutor(etlConf, null, registry, jobInstance.getDefId(), jobInstance.getInstId(), pluginId);
        try {
            etlPluginExecutor.doStart();
            etlPluginExecutor.getAttemptCount();

            reporter = ConsoleReporter.forRegistry(registry).build();
            SortedMap<String, Counter> counters = registry.getCounters();
            reporter.start(20, TimeUnit.SECONDS);

            while (!etlPluginExecutor.isTaskFinished()) {
                try {
                    TimeUnit.NANOSECONDS.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {

        } finally {
            log.info("==============定义{}ETL任务执行完成==============", etlInstId);
            reporter.report();
            reporter.close();
            etlPluginExecutor.shutdown();
        }
    }

    @Override
    public void destroy() {

    }

    @Override
    public JobHandler clone() {
        try {
            return (JobHandler) super.clone();
        } catch (CloneNotSupportedException e) {
            throw new SchedulerException("调度服务克隆异常！", e);
        }
    }

    private void initService() {
        if (environment == null) {
            environment = (Environment) AppUtil.getBean(Environment.class);
        }
    }
}
